package com.klm.easymq.service.impl;

import com.klm.easymq.service.RocketMQProducerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

/**
 * RocketMQ消息生产者服务实现
 */
@Slf4j
@Service
public class RocketMQProducerServiceImpl implements RocketMQProducerService {
    
    @Autowired
    private RocketMQTemplate rocketMqTemplate;
    
    @Override
    public boolean sendSyncMessage(String topic, String message) {
        return sendSyncMessage(topic, message, "", "");
    }
    
    @Override
    public boolean sendSyncMessage(String topic, String message, String tag, String messageKey) {
        try {
            org.springframework.messaging.Message<String> msg = MessageBuilder.withPayload(message)
                    .setHeader("tag", tag)
                    .setHeader("key", messageKey)
                    .build();
            
            SendResult sendResult = rocketMqTemplate.syncSend(topic, msg);
            log.info("同步消息发送成功: topic={}, messageId={}, tag={}, key={}", 
                    topic, sendResult.getMsgId(), tag, messageKey);
            return true;
        } catch (Exception e) {
            log.error("同步消息发送失败: topic={}, message={}, tag={}, key={}", 
                    topic, message, tag, messageKey, e);
            return false;
        }
    }
    
    @Override
    public void sendAsyncMessage(String topic, String message) {
        sendAsyncMessage(topic, message, "", "");
    }
    
    @Override
    public void sendAsyncMessage(String topic, String message, String tag, String messageKey) {
        try {
            org.springframework.messaging.Message<String> msg = MessageBuilder.withPayload(message)
                    .setHeader("tag", tag)
                    .setHeader("key", messageKey)
                    .build();
            
            rocketMqTemplate.asyncSend(topic, msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("异步消息发送成功: topic={}, messageId={}, tag={}, key={}", 
                            topic, sendResult.getMsgId(), tag, messageKey);
                }
                
                @Override
                public void onException(Throwable throwable) {
                    log.error("异步消息发送失败: topic={}, message={}, tag={}, key={}", 
                            topic, message, tag, messageKey, throwable);
                }
            });
        } catch (Exception e) {
            log.error("异步消息发送异常: topic={}, message={}, tag={}, key={}", 
                    topic, message, tag, messageKey, e);
        }
    }
    
    @Override
    public boolean sendDelayMessage(String topic, String message, int delayLevel) {
        try {
            org.springframework.messaging.Message<String> msg = MessageBuilder.withPayload(message).build();
            SendResult sendResult = rocketMqTemplate.syncSend(topic, msg, 3000, delayLevel);
            log.info("延迟消息发送成功: topic={}, messageId={}, delayLevel={}", 
                    topic, sendResult.getMsgId(), delayLevel);
            return true;
        } catch (Exception e) {
            log.error("延迟消息发送失败: topic={}, message={}, delayLevel={}", topic, message, delayLevel, e);
            return false;
        }
    }
    
    @Override
    public boolean sendOrderedMessage(String topic, String message, String hashKey) {
        try {
            SendResult sendResult = rocketMqTemplate.syncSendOrderly(topic, message, hashKey);
            log.info("顺序消息发送成功: topic={}, messageId={}, hashKey={}", 
                    topic, sendResult.getMsgId(), hashKey);
            return true;
        } catch (Exception e) {
            log.error("顺序消息发送失败: topic={}, message={}, hashKey={}", topic, message, hashKey, e);
            return false;
        }
    }
    
    @Override
    public void sendOnewayMessage(String topic, String message, String tag, String messageKey) {
        try {
            org.springframework.messaging.Message<String> msg = MessageBuilder.withPayload(message)
                    .setHeader("tag", tag)
                    .setHeader("key", messageKey)
                    .build();
            
            rocketMqTemplate.sendOneWay(topic, msg);
            log.info("单向消息发送成功: topic={}, tag={}, key={}", topic, tag, messageKey);
        } catch (Exception e) {
            log.error("单向消息发送失败: topic={}, message={}, tag={}, key={}", 
                    topic, message, tag, messageKey, e);
        }
    }
} 